package com.echo.im.server.netty.processor.impl;


import com.echo.im.common.enums.IMCmdType;
import com.echo.im.common.model.IMRecvInfo;
import com.echo.im.common.model.IMSendInfo;
import com.echo.im.common.model.IMUserInfo;
import com.echo.im.common.model.mq.OfflineMsgMqInfo;
import com.echo.im.common.vo.message.PrivateMessageVO;
import com.echo.im.mq.event.producer.EventPublisher;
import com.echo.im.mq.event.model.OfflineMessageEvent;
import com.echo.im.server.ack.MessageAckManager;
import com.echo.im.server.netty.context.UserChannelMap;
import com.echo.im.server.netty.processor.AbstractMessageProcessor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

/**
 * 1. 浏览器发送 【 `content` `type` `recvId` 】到 `im-platform`
 * 2. `im-platform` 将此消息异步入库
 * 3. `im-platfrom` 请求 `im-client` 进行消息推送
 * 4. `im-client` 通过 `Redis` 查询后获取到接受者连接的 `im-server` 对应的 RabbitMQ routeKey，发送消息到此 routeKey
 *    1. 如果用户不在线，则消息入离线消息队列（如果redis上没有缓存user-serverId键值对，说明用户下线），**至此消息发送成功**，给前端发送 ack（HTTP请求不用管，因为一个HTTP请求已经包含a一来一回了）
 * 5. `im-server` 监听到此消息，进行发送
 *    1. 去 `ConcurrentHashMap` 中取出该用户的 Channel，通过此 Channel 发送消息
 *    2. 如果不存在 `Channel`，则证明用户已经离线，则消息入离线消息队列，**至此消息发送成功**
 *    3. 发送消息，并存入 ack 等待队列，等待接受者的 ack
 *    4. 超时未等到 ack，则重复 **步骤 1** （此处实践上来说失败次数会很少，可忽略不计，没必要过度设计）
 *    5. 收到 ack，则从 ack 等待队列中删除消息，**至此消息发送成功**
 * @author echo
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class PrivateMessageProcessor extends AbstractMessageProcessor<IMRecvInfo<PrivateMessageVO>> {

    private final MessageAckManager messageAckManager;
    private final EventPublisher eventPublisher;
    private final OfflineMessageEvent offlineMessageEvent;
    @Override
    public void process(ChannelHandlerContext ctx, IMRecvInfo<PrivateMessageVO> data) {

    }

    @Override
    public void process(IMRecvInfo<PrivateMessageVO> recvInfo) {
        IMUserInfo sender = recvInfo.getSender();
        List<IMUserInfo> receivers = recvInfo.getReceivers();
        for (IMUserInfo receiver : receivers) {
            log.info("接收到私聊消息，发送者:{},接收者:{}，内容:{}", sender.getId(), receiver.getId(), recvInfo.getData());
            try {
                Channel channel = UserChannelMap.getChannel(receiver.getId(), receiver.getTerminal());

                //Channel为null，则证明用户不在线，则消息入离线消息队列
                if (Objects.isNull(channel)){
                    OfflineMsgMqInfo offlineMsgMqInfo = new OfflineMsgMqInfo(receiver.getId(), recvInfo.getData().getId(),OfflineMsgMqInfo.OfflineMsgEnum.PRIVATE_OFFLINE_MESSAGE);
                    eventPublisher.publish(offlineMessageEvent.topic(),offlineMessageEvent.buildEventMessage(offlineMsgMqInfo));
                    return;
                }

                //推送消息
                IMSendInfo<Object> sendInfo = new IMSendInfo<>();
                sendInfo.setCmd(IMCmdType.PRIVATE_MESSAGE.code());
                sendInfo.setData(recvInfo.getData());
                channel.writeAndFlush(sendInfo);

                //并存入 ack 等待队列，等待接受者的 ack，如果超时未收到 ack，直接断开连接，让用户拉取离线消息
                //ack队列里面只应该存储用户和消息id，只要用户的所有在线终端，有一个收到了消息，就够了，如果超过3s未收到ack，直接断开连接，并将消息加入离线队列
                messageAckManager.addPendingAck(recvInfo.getData().getId(), receiver.getId(), channel.id().asLongText(), OfflineMsgMqInfo.OfflineMsgEnum.PRIVATE_OFFLINE_MESSAGE);

            } catch (Exception e) {
                //  直接将消息加入离线消息
                OfflineMsgMqInfo offlineMsgMqInfo = new OfflineMsgMqInfo(receiver.getId(), recvInfo.getData().getId(),OfflineMsgMqInfo.OfflineMsgEnum.PRIVATE_OFFLINE_MESSAGE);
                eventPublisher.publish(offlineMessageEvent.topic(),offlineMessageEvent.buildEventMessage(offlineMsgMqInfo));
                log.error("发送异常，发送者:{},接收者:{}，内容:{}", sender.getId(), receiver.getId(), recvInfo.getData(), e);
            }
        }
    }

    @Override
    public IMCmdType getCmdType() {
        return IMCmdType.PRIVATE_MESSAGE;
    }

    @Override
    public IMRecvInfo transForm(Object o) {
        return null;
    }
}
